package hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;

import static org.apache.hadoop.hbase.filter.CompareFilter.CompareOp.EQUAL;


public class HBaseFilter {

    Connection conn;
    Admin admin;
    Table sT;


    @Before
    public void create() throws IOException {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master,node1,node2");
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
        sT = conn.getTable(TableName.valueOf("students"));
    }

    @Test
    //RegexStringComparator+ValueFilter
    //ValueFilter会将比较器作用到所有的列值上，相当于作用到每一个cell上
    //把符合比较器规则的cell过滤出来，不符合的直接丢弃
    public void Filter1() throws IOException {

        Scan scan = new Scan();

        //^以什么为开头，.*表示全部，$表示已什么为结尾
        RegexStringComparator rsc = new RegexStringComparator("^文科.*班$");
        ValueFilter valueFilter = new ValueFilter(EQUAL, rsc);
        scan.setFilter(valueFilter);
        ResultScanner rs = sT.getScanner(scan);
        for (Result r : rs) {

            for (Cell cell : r.listCells()) {
                String values = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println(values);
            }
//            String id = Bytes.toString(r.getRow());
//            String name = Bytes.toString(r.getValue("info".getBytes(), "name".getBytes()));
//            String age = Bytes.toString(r.getValue("info".getBytes(), "age".getBytes()));
//            String sex = Bytes.toString(r.getValue("info".getBytes(), "sex".getBytes()));
//            String clazz = Bytes.toString(r.getValue("info".getBytes(), "clazz".getBytes()));
//            System.out.println(id+","+name+","+age+","+sex+","+clazz);
        }


    }

    @Test
    // RegexStringComparator比较器
    // SingleColumnValueFilter过滤器，对单列进行过滤，
    // 注意和SingleColumnValueExcludeFilter区分，这也是过滤列，但不输出此列
    public void Filter2() throws IOException {
        Scan scan = new Scan();


        SingleColumnValueFilter scv =
                new SingleColumnValueFilter("info".getBytes(),"clazz".getBytes(),
                        CompareFilter.CompareOp.EQUAL,
                        new RegexStringComparator("^文科.*班$"));

        scan.setFilter(scv);
        ResultScanner rs = sT.getScanner(scan);
        for (Result r : rs) {
            String s = Bytes.toString(r.getRow());
            for (Cell cell : r.listCells()) {

                String values = Bytes.toString(CellUtil.cloneValue(cell));
                s=s+" "+values;

            }
            System.out.println(s);

        }
    }

    @Test
    // SubstringComparator 判断提供的子串是否出现在中
    // QualifierFilter,列过滤器，根据列名过滤出符合条件的列
    public void Filter3() throws IOException {

        //筛选出age，name，clazz列
        SubstringComparator a = new SubstringComparator("a");
        QualifierFilter qualifierFilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, a);

        //直接调用我下面写的方法
        printScannerWithFilter(qualifierFilter);
    }

    //这里可以写一个方法把重复的代码进行处理
    public void printScannerWithFilter(Filter filter) throws IOException {
        Scan scan = new Scan();
        scan.setFilter(filter);
        ResultScanner rs = sT.getScanner(scan);
        for (Result r : rs) {
            String s = Bytes.toString(r.getRow());
            for (Cell cell : r.listCells()) {
                String values = Bytes.toString(CellUtil.cloneValue(cell));
                s+=" "+values+" ";
            }
            System.out.println(s);
        }
    }

    @Test
    //BinaryPrefixComparator比较器，按字节索引顺序比较指定字节数组，只是比较左端前缀的数据是否相同
    //RowFilter 行键过滤器
    public void Filter4() throws IOException {
        BinaryPrefixComparator bpc = new BinaryPrefixComparator("150010098".getBytes());
        RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, bpc);
        printScannerWithFilter(rowFilter);
    }


    //专用过滤器：rowkey前置过滤器PrefixFliter
    //相当于BinaryPrefixComparator+RowFilter
    @Test
    public  void Filter5() throws IOException {
        PrefixFilter prefixFilter = new PrefixFilter("150010098".getBytes());
        printScannerWithFilter(prefixFilter);
    }

    @Test
    //过滤age>=23,sex=男，ROWKEY以150010010开头所有学生数据
    public void Filter6() throws IOException {
        SingleColumnValueFilter age =
                new SingleColumnValueFilter(
                        "info".getBytes(),
                        "age".getBytes(),
                        CompareFilter.CompareOp.GREATER_OR_EQUAL,
                        "23".getBytes()
                );
        SingleColumnValueFilter sex =
                new SingleColumnValueFilter(
                        "info".getBytes(),
                        "sex".getBytes(),
                        CompareFilter.CompareOp.EQUAL, "男".getBytes()
                );

        PrefixFilter prefixFilter = new PrefixFilter("150010010".getBytes());
        //创建一个filter数组存储多个过滤器
        FilterList filterList = new FilterList();
        filterList.addFilter(age);
        filterList.addFilter(sex);
        filterList.addFilter(prefixFilter);

        printScannerWithFilter(filterList);
    }

    @After
    public void  close() throws IOException {
        conn.close();
    }
}
